package net.bwie.dws;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;

public class dws草稿 {
    public static void main(String[] args) {
        EnvironmentSettings build = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tableEnv = TableEnvironment.create(build);

        Configuration configuration = tableEnv.getConfig().getConfiguration();
        configuration.setString("table.local-time-zone", "Asia/Shanghai");
        configuration.setString("table.exec.resource.default-parallelism", "1");
        configuration.setString("table.exec.state.ttl", "5 s");
        configuration.setString("execution.checkpointing.interval", "5 s");


        tableEnv.executeSql("CREATE TABLE dwd_retreat_behavior (\n" +
                        //订单id
                        "order_id STRING ,\n" +
                        //  商品ID
                        "product_id STRING ,\n" +
                        // 订单支付时间
                        "pay_time STRING ,\n" +
//                    （订单总金额）
                        "order_amount DOUBLE,\n" +

                        // 退款id
                        "refund_id STRING ,\n" +
                        // 退款申请时间
                        "apply_time TIMESTAMP(3) ,\n" +
                        //  退款场景:“退货退款”“已收货仅退款”“未收货退款”“未发货仅退款”
                        "refund_scene STRING ,\n" +
                        // 实际退款金额
                        "refund_amount DOUBLE ,\n" +
                        // 退款状态
                        "refund_status STRING,\n" +

                        //消费者行为id
                        "behavior_id STRING ,\n" +
                        //消费者id
                        "user_id STRING ,\n" +
                        //行为类型
                        "behavior_type STRING, \n" +

                        " time_ltz AS TO_TIMESTAMP_LTZ(UNIX_TIMESTAMP(pay_time),0),\n" +
                        "WATERMARK FOR time_ltz AS time_ltz - INTERVAL '5' DAY\n" +
                        ")WITH(\n" +
                        "'connector' = 'kafka',\n" +
                        "'topic' = 'dwd_retreat_behavior',\n" +
                        "'properties.bootstrap.servers' = 'node101:9092',\n" +
                        "'properties.group.id' = 'dwd_retreat_behavior-id04',\n" +
                        "'scan.startup.mode' = 'earliest-offset',\n" +
                        "'format' = 'json'\n" +
                        ")"
        );



//            tableEnv.executeSql("SELECT * FROM  dwd_retreat_behavior").print();


            /*
                    //订单id
                    "order_id,\n" +
                    // 实际退款金额
                    "refund_amount,\n" +
                    //（订单总金额）
                    "order_amount,\n" +
                    // 退款id
                    "refund_id,\n"+
                    //退款状态
                    "refund_status\n" +
             */

        /*
            "SELECT \n" +
                            "product_id,\n" +
                            "refund_scene,\n" +
                            "user_count,\n" +
                            "order_count/order_sum,\n" +
                            "amout_sum/order_sum" +
                            "FROM(" +

                                "SELECT \n" +
                                // 商品id
                                "product_id,\n" +
                                // 退款场景
                                "refund_scene,\n" +

                                // 退款人数
                                "count(user_id) as user_count,\n"+

                                // 各个退款场景订单数
                                "COUNT(order_id) as order_count,\n"+

                                // 各个退款场景退款订单数
                                // 使用sum if 聚合函数 求出退款订单数
                                "sum(" +
                                "if(refund_status= '申请成功',1,0 )" +
                                ") as order_sum,\n" +

                                //total_amount（总金额）
                                "ROUND(SUM(order_amount),2) as amout_sum,\n" +
                                //refund_amount（退款金额）
                                "ROUND(" +
                                "sum(" +
                                "if(refund_status= '申请成功',refund_amount,0 )" +
                                ")"+
                                ",2) as order_sum\n" +


                                "FROM dwd_retreat_behavior\n" +
                                "GROUP BY product_id,refund_scene"+

                            ")a1"+
                            ""

         */
        /*
              "        -- 订单退款率 = 退款子订单数 / 总子订单数\n" +
                "        CASE WHEN COUNT(DISTINCT   order_id) = 0 THEN 0\n" +
                "             ELSE COUNT(DISTINCT  order_id) / COUNT(DISTINCT   order_id)\n" +
                "            END AS order_refund_rate,\n" +

                    "        -- 金额退款率 = 退款金额 / 总金额\n" +
                "        CASE WHEN SUM(  order_amount) = 0 THEN 0\n" +
                "             ELSE SUM( refund_amount) / SUM(  order_amount)\n" +
                "            END AS amount_refund_rate\n" +
         */

        tableEnv.executeSql(" SELECT\n" +
                " window_start ,\n" +
                " window_end,\n" +
                // 退款场景
                " refund_scene,\n" +
                " COUNT(user_id) AS refund_user_count, -- 退款人数\n" +

                " sum(if(refund_id IS NOT NULL,1,0)) AS refund_sub_order_count, -- 退款子订单数\n" +

                " round(SUM(refund_amount),2) AS total_refund_amount -- 退款总金额\n" +
                " FROM TABLE(\n" +
                " -- 添加 1天 的 滚动窗口\n" +
                "  TUMBLE(TABLE dwd_retreat_behavior, DESCRIPTOR(time_ltz), INTERVAL '1' day)\n" +
                "                        )\n" +
//                "-- 筛选有效退款场景（四类场景）\n" +
                "    WHERE  refund_scene IN ('退货退款', '已收货仅退款', '未收货退款', '未发货仅退款')\n" +
                "    GROUP BY\n" +
                "        window_start ,\n" +
                "        window_end,\n" +
                "         refund_scene\n" +
                "         ").print();



/*
        tableEnv.executeSql(
                "SELECT \n" +
                        " window_start,\n" +
                        " window_end,\n" +
                        " product_id,\n" +
                        " refund_scene,\n" +
                        " user_count,\n" +
                        " CAST(order_count AS DOUBLE) / CAST(refund_count AS DOUBLE) AS refund_rate,\n" +
                        " CAST(total_amount AS DOUBLE) / CAST(refund_count AS DOUBLE) AS avg_refund_amount\n" +
                        "FROM (SELECT \n" +
                        "  window_start,\n" +
                        "  window_end,\n" +
                        "  product_id,\n" +
                        "  refund_scene,\n" +
                        "  COUNT(DISTINCT user_id) as user_count,\n" +
                        "  COUNT(order_id) as order_count,\n" +
                        "  SUM(CASE WHEN refund_status = '申请成功' THEN 1 ELSE 0 END) as refund_count,\n" +
                        "  ROUND(SUM(order_amount), 2) as total_amount,\n" +
                        "  ROUND(SUM(CASE WHEN refund_status = '申请成功' THEN refund_amount ELSE 0 END), 2) as refund_amount_sum\n" +
                        "FROM TABLE(\n" +
                        // 添加 1天 的 滚动窗口
                        "  TUMBLE(TABLE dwd_retreat_behavior, DESCRIPTOR(time_ltz), INTERVAL '1' day)\n" +
                        ")\n" +
                        "      GROUP BY window_start,window_end,product_id, refund_scene\n" +
                        ") a1 "
        ).print();
 */




        TableResult tableResult = tableEnv.executeSql(
                "SELECT\n" +
                        "window_start,\n" +
                        "window_end,\n" +
                        "refund_time_type,\n" +
                        "count(order_id) as order_num\n" +
                        "FROM (\n" +
                        "SELECT \n" +
                        "window_start,\n" +
                        "window_end,\n" +
                        "CASE \n" +
                        "   WHEN diff_minutes<=30 THEN '支付30分钟内'\n" +
                        "   WHEN diff_minutes>30 and diff_hours<=24 THEN '30分钟-24小时'\n" +
                        "   ELSE '24小时以上'\n" +
                        "END AS refund_time_type,\n" +
                        "order_id\n" +
                        "FROM (\n" +
                        "SELECT \n" +
                        "window_start\n" +
                        ",window_end\n" +
                        ",order_id\n" +
                        ",time_ltz\n" +
                        ",apply_time\n" +
                        ",TIMESTAMPDIFF(MINUTE, time_ltz, CAST(apply_time AS TIMESTAMP_LTZ(3))) AS diff_minutes\n" +
                        ",TIMESTAMPDIFF(HOUR, time_ltz, CAST(apply_time AS TIMESTAMP_LTZ(3))) AS diff_hours\n" +
                        "FROM TABLE(\n" +
                        // 添加 1天 的 滚动窗口
                        "  TUMBLE(TABLE dwd_retreat_behavior, DESCRIPTOR(time_ltz), INTERVAL '1' day)\n" +
                        ")\n" +
                        "WHERE time_ltz IS NOT NULL AND apply_time IS NOT NULL\n" +
                        ")\n" +
                        ")\n" +
                        "GROUP BY window_start,window_end,refund_time_type\n"
        );
        tableResult.print();


    }
}
